from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.ml.stat import Correlation
from pyspark.ml.feature import VectorAssembler

# 创建 SparkSession
# SparkSession.builder：用于构建 SparkSession 对象。
# .appName("AliEcommerceSearchAnalysis")：设置 Spark 应用程序的名称。
# .getOrCreate()：如果已经存在一个 SparkSession，则返回该对象；否则，创建一个新的 SparkSession。
spark = SparkSession.builder \
    .appName("AliEcommerceSearchAnalysis") \
    .getOrCreate()

# 加载数据
# spark.read.csv：从 CSV 文件中读取数据。
# header=True：表示 CSV 文件包含表头。
# inferSchema=True：表示自动推断数据类型。
products_df = spark.read.csv("../mock_data/products.csv", header=True, inferSchema=True)
users_df = spark.read.csv("../mock_data/users.csv", header=True, inferSchema=True)
keywords_df = spark.read.csv("../mock_data/search_keywords.csv", header=True, inferSchema=True)
search_records_df = spark.read.csv("../mock_data/search_records.csv", header=True, inferSchema=True)
reviews_df = spark.read.csv("../mock_data/product_reviews.csv", header=True, inferSchema=True)

# 找出搜索量最高的前 10 个关键词
# select("keyword_text", "search_volume")：选择 keyword_text 和 search_volume 两列。
# orderBy(col("search_volume").desc())：按照 search_volume 列降序排序。
# limit(10)：选取前 10 条记录。
# show()：显示 DataFrame 的内容。
top_keywords = keywords_df.select("keyword_text", "search_volume") \
    .orderBy(col("search_volume").desc()) \
    .limit(10)

print("热门搜索关键词：")
top_keywords.show()

# 统计不同搜索频率的用户数量
# groupBy("search_frequency")：按照 search_frequency 列进行分组。
# count()：统计每个分组的记录数。
# orderBy("search_frequency")：按照 search_frequency 列升序排序。
search_frequency_distribution = users_df.groupBy("search_frequency") \
    .count() \
    .orderBy("search_frequency")

print("用户搜索频率分布：")
search_frequency_distribution.show()

# 在连接前重命名 reviews_df 中的 rating 列
reviews_df.show()
reviews_df = reviews_df.withColumnRenamed("rating", "review_rating")
reviews_df.show()

# 关联商品表和评价表
# withColumnRenamed("rating", "review_rating")：将 reviews_df 中的 rating 列重命名为 review_rating，避免与其他表中的列名冲突。
# join(reviews_df, on="product_id", how="inner")：将 products_df 和 reviews_df 通过 product_id 列进行内连接。
product_review_df = products_df.join(reviews_df, on="product_id", how="inner")

# 选择需要的列，使用重命名后的列名
selected_df = product_review_df.select("sales_volume", "review_rating")

# 将列转换为向量
# select("sales_volume", "review_rating")：选择 sales_volume 和 review_rating 两列。
# VectorAssembler：将 sales_volume 和 review_rating 两列合并为一个向量列 features。
# transform(selected_df)：对 selected_df 应用 VectorAssembler 转换。
assembler = VectorAssembler(inputCols=["sales_volume", "review_rating"], outputCol="features")
vector_df = assembler.transform(selected_df).select("features")
vector_df.show()

# 计算皮尔逊相关系数
# Correlation.corr(vector_df, "features")：计算 vector_df 中 features 列的相关性矩阵。
# collect()[0][0]：获取相关性矩阵的第一行第一列元素。
# corr_matrix[0, 1]：获取相关性矩阵中 sales_volume 和 review_rating 之间的皮尔逊相关系数。
corr_matrix = Correlation.corr(vector_df, "features").collect()[0][0]
pearson_corr = corr_matrix[0, 1]
print(f"商品销量与评价评分的皮尔逊相关系数：{pearson_corr}")

# 停止 SparkSession
spark.stop()






















